package com.atguigu.chapter07.b_watermarket;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @ClassName: Flink01_Tumbling_Window
 * @Description:
 * @Author: kele
 * @Date: 2021/4/6 20:06
 *
 * 侧输出流测试
 *
 * 使用侧输出流将不同的key对应的输出到不同的流中,只要有上下文Context就可以
 *
 **/
public class Flink13_WaterMarket_Slide_output_2 {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        SingleOutputStreamOperator<String> main = env.socketTextStream("hadoop162", 8888)
                .map(datas -> {

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                .process(new ProcessFunction<WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value,
                                               Context ctx,
                                               Collector<String> out) throws Exception {

                        if (value.getId().equals("a")) {
                            ctx.output(new OutputTag<WaterSensor>("a") {
                            }, value);
                        } else if (value.getId().equals("b")) {
                            ctx.output(new OutputTag<WaterSensor>("b") {
                            }, value);
                        } else {
                            out.collect(value.toString());
                        }
                    }
                });

        main.print("main");
        main.getSideOutput(new OutputTag<WaterSensor>("a"){}).print("a");
        main.getSideOutput(new OutputTag<WaterSensor>("b"){}).print("b");


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
